package com.oildatacollection.datareceive;

import com.oildatacollection.datareceive.datastore.StoreManager;
import com.oildatacollection.datareceive.datastore.StoreManagerInfluxImpl;
import com.oildatacollection.message.RecvMessage;
import com.oildatacollection.util.YamlLoader;
import javafx.util.Pair;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class RecvQueueImpl implements RecvQueue {
    private LinkedBlockingQueue<RecvMessage> lbq;
    private final AtomicBoolean stoped;
    private final StoreManager storeManager;
    private Thread recvThread;

    public RecvQueueImpl() {
        lbq = null;
        stoped = new AtomicBoolean(true);
        storeManager = new StoreManagerInfluxImpl();
        recvThread = null;
    }

    private void initConfig() {
        try {
            YamlLoader yamlLoader = new YamlLoader("app-config/application-server.yml");
            Map<String, Object> ymlObj = (Map<String, Object>) yamlLoader.get("dataRecv");
            int capacity = (int) ymlObj.get("queueCapacity");
            lbq = new LinkedBlockingQueue<>(capacity);
        } catch (RuntimeException e) {
            lbq = new LinkedBlockingQueue<>(100);
            log.error("RecvQueue: 找不到配置文件", e);
        }
        storeManager.connectDB();
        log.trace("接收队列初始化完毕");
    }

    /**
     * 接收线程类
     */
    private class QueThread implements Runnable {
        @Override
        public void run() {
            while (!stoped.get()) {
                try {
                    // 获取消息
                    RecvMessage msg = lbq.take();

                    msg.resetIter();
                    String msgCont = "";
                    while (msg.hasNext()) {
                        Pair<String, Object> item = msg.getNext();
                        msgCont += item.getKey() + ":" + item.getValue() + ";";
                    }
                    log.trace(msgCont);

                    // 获取空消息时，就是队列停止的时候
                    if (msg.IsEmpty()) {
                        break;
                    }

                    // 将消息插入到时序数据库中
                    storeManager.insertMessageToDB(msg);
                } catch (InterruptedException e) {
                    log.error("ReceverQueue 错误: ", e);
                }
            }
            log.trace("接收队列终止运行");
        }
    }

    @Override
    public void sendToQueue(RecvMessage rcMsg) {
        try {
            lbq.put(rcMsg);
        } catch (InterruptedException e) {
            log.error("sendToQueue错误", e);
        }
    }

    @Override
    public void start() {
        initConfig();
        stoped.set(false);
        // 开启队列线程
        recvThread = new Thread(new QueThread(), "RECEVER_QUE_THREAD");
        recvThread.start();
        log.trace("接收队列模块启动成功");
    }

    @Override
    public void stop() {
        stoped.set(true);
        sendToQueue(new RecvMessage());
        try {
            recvThread.join();
        } catch (InterruptedException e) {
            log.error("RecvQueue::stop", e);
        }
    }
}
